use std::{path::PathBuf, sync::Arc};
use bytes::{Buf, BytesMut};
use parking_lot::{RwLock};

use crate::{
    errors::{DbErrors, Result},
    fio::{self, new_io_manager},
};

use super::log_record::{max_log_record_header_size, LogRecord, LogRecordType, ReadLogRecord};

use prost::{decode_length_delimiter, encode_length_delimiter, length_delimiter_len};


// 文件后缀
pub const DATA_FILE_NAME_SUFFIX: &str = ".data";

pub struct DataFile {
    file_id: Arc<RwLock<u32>>,      // 数据文件id
    write_off: Arc<RwLock<u64>>,    // 当前写偏移，记录该数据文件写到哪个位置了
    io_manager: Box<dyn fio::IoManager>,    // IO 管理接口
}


impl DataFile {

    /// 打开新的文件句柄
    pub fn new(dir_path: PathBuf, file_id: u32) -> Result<DataFile> {
        let file_name = get_data_file_name(dir_path, file_id);
        let io_manager = new_io_manager(file_name)?;
        Ok(
            DataFile { 
                file_id: Arc::new(RwLock::new(file_id)), 
                write_off: Arc::new(RwLock::new(0)), 
                io_manager: Box::new(io_manager),
            }
        )
    }

    pub fn get_write_off(&self) -> u64 {
        let read = self.write_off.read();
        *read
    }

    /// 
    pub fn set_write_off(&self, offset: u64) {
        let mut wr = self.write_off.write();
        *wr = offset;
    }

    /// 获取文件 id
    pub fn get_file_id(&self) -> u32 {
        let file_id = self.file_id.read();
        *file_id
    }

    /// 读取记录
    pub fn read_log_record(&self, offset: u64) -> Result<ReadLogRecord> {
        // 先读取 header 部分数据
        let mut header_buf = BytesMut::zeroed(max_log_record_header_size());

        self.io_manager.read(&mut header_buf, offset)?;

        let rec_type = header_buf.get_u8();


        let key_size = decode_length_delimiter(&mut header_buf).unwrap();
        let val_size = decode_length_delimiter(&mut header_buf).unwrap();
        

        // time is over
        if key_size == 0 && val_size == 0 {
            return Err(DbErrors::ReadDataFileEOF);
        }
        // 获取实际的 header 大小
        let actual_header_size = length_delimiter_len(key_size) + length_delimiter_len(val_size);

        // 读取实际的 key 和 value，最后的 4 个字节是 crc 校验值
        let mut kv_buf = BytesMut::zeroed(key_size + val_size + 4);   
        self.io_manager.read(&mut kv_buf, offset + actual_header_size as u64)? ;

        let log_record = LogRecord{
            key: kv_buf.get(..key_size).unwrap().to_vec(),
            value: kv_buf.get(key_size..kv_buf.len() - 4).unwrap().to_vec(),
            rec_type: rec_type.into(),
        };

        kv_buf.advance(key_size + val_size);

        if kv_buf.get_u32() != log_record.get_crc() {
            return Err(DbErrors::InvalidLogRecordCrc) ;
        }


        Ok(ReadLogRecord { 
            record: log_record, 
            size: actual_header_size + key_size + val_size + 4 
        })
    }

    /// 写入
    pub fn write(&self, buf: &[u8]) -> Result<usize> {
        let n_bytes = self.io_manager.write(buf)?;

        let mut write_off = self.write_off.write();
        *write_off += n_bytes as u64;
        
        Ok(n_bytes)
    }
    
    /// 同步
    pub fn sync(&self) -> Result<()> {
        self.io_manager.sync()
    }

}

/// 获取文件名称
fn get_data_file_name(dir_path: PathBuf, file_id: u32) -> PathBuf {
    let file_name = std::format!("{:09}", file_id) + DATA_FILE_NAME_SUFFIX;
    dir_path.join(file_name)
}


#[cfg(test)]
mod test {
    
    use super::*;

    #[test]
    fn test() {
        let dir_path = std::env::temp_dir();
        let data_file_res1 = DataFile::new(dir_path.clone(), 0);
        assert!(data_file_res1.is_ok());

        let data_file1 = data_file_res1.unwrap();
        assert_eq!(data_file1.get_file_id(), 0);

    }

    #[test]
    fn test_data_file_write() {
        let dir_path = std::env::temp_dir();
        let data_file_res1 = DataFile::new(dir_path.clone(), 100);
        assert!(data_file_res1.is_ok());
        let data_file1 = data_file_res1.unwrap();
        assert_eq!(data_file1.get_file_id(), 100);
    
        let write_res1 = data_file1.write("aaa".as_bytes());
        assert!(write_res1.is_ok());
        assert_eq!(write_res1.unwrap(), 3_usize);

    }


    #[test]
    fn test_data_file_sync() {
        let dir_path = std::env::temp_dir();
        let data_file_res1 = DataFile::new(dir_path.clone(), 200);
        assert!(data_file_res1.is_ok());
        let data_file1 = data_file_res1.unwrap();
        assert_eq!(data_file1.get_file_id(), 200);

        let sync_res = data_file1.sync();
        assert!(sync_res.is_ok());
    }


}

